package com.tmsb.sink;

import com.tmsb.pojo.ESStockAvailable;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.MyMath;
import com.tmsb.utils.RedisCluster;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author wuxiangwu
 * @Date 2020/8/13 9:29
 * @Description 统计每日释放绩效积分
 * @Version 1.0
 */
public class AvailableBalanceSink extends RichSinkFunction<ESStockAvailable> implements CheckpointedFunction {

    private ListState<Double> allAmountValueState;
    private ListState<Long> dateState;
    private Double allAmountValue;
    private Long dateValue;
    private JedisCluster jedisCluster;
    private SimpleDateFormat sdf;

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        allAmountValueState.clear();
        dateState.clear();
        allAmountValueState.add(allAmountValue);
        dateState.add(dateValue);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        allAmountValue = 0.0;
        dateValue = 0L;
         ListStateDescriptor<Double> allAmountStateDes = new ListStateDescriptor<>("allAmountState", TypeInformation.of(new TypeHint<Double>() {
        }));
         ListStateDescriptor<Long> dateStateDes = new ListStateDescriptor<>("dateState", TypeInformation.of(new TypeHint<Long>() {
        }));

        allAmountValueState = context.getOperatorStateStore().getListState(allAmountStateDes);
        dateState =  context.getOperatorStateStore().getListState(dateStateDes);
        if (context.isRestored()) {
            allAmountValue = allAmountValueState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        jedisCluster = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");
    }

    @Override
    public void invoke(ESStockAvailable value, Context context) throws Exception {
        //TODO 每天0时清空状态
         Long last = value.getLast();
         if(last >= dateValue){
             allAmountValue = 0.0;
             long nowMillis = System.currentTimeMillis();
             dateValue = nowMillis - (nowMillis + 60 * 60 * 8 * 1000) % (1000 * 60 * 60 * 24L) +(1000 * 60 * 60 * 24L);
        //     System.out.println("第二天零点的时间戳="+dateValue);
         }else if(last < Util.getZeroPointTimestamps()){
        //     System.out.println("今天零点的时间戳="+Util.getZeroPointTimestamps());
             return;
         }

         //TODO 每日锁仓释放量
         Double amount = value.getAmount();
         String add = MyMath.add(allAmountValue + "", amount + "");
        allAmountValue = Double.parseDouble(add);
        String format = sdf.format(new Date(last));
        jedisCluster.hset("es:available:amount",format,allAmountValue+"");
    }
}
